package com.example.java.cep;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

/**
 * flink CEP开发流程
 * 1. DataSource 中的数据转换为 DataStream；watermark、keyby
 * 2. 定义 Pattern，并将 DataStream 和 Pattern 组合转换为 PatternStream；
 * 3. PatternStream 经过 select、process 等算子转换为 DataStream；
 * 4. 再次转换的 DataStream 经过处理后，sink 到目标库。
 * 目的:检测用户恶意登录
 */

public class LoginTooManyDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        DataStreamSource<CEPLoginBean> data = env.fromElements(
                new CEPLoginBean(1L, "fail", 1597905234000L),
                new CEPLoginBean(1L, "success", 1597905235000L),
                new CEPLoginBean(2L, "fail", 1597905236000L),
                new CEPLoginBean(2L, "fail", 1597905237000L),
                new CEPLoginBean(2L, "fail", 1597905238000L),
                new CEPLoginBean(3L, "fail", 1597905239000L),
                new CEPLoginBean(3L, "success", 1597905240000L)
        );

        // 为数据添加水印时间
        SingleOutputStreamOperator<CEPLoginBean> watermarks = data
                .assignTimestampsAndWatermarks(new WatermarkStrategy<CEPLoginBean>() {
                    @Override
                    public WatermarkGenerator<CEPLoginBean> createWatermarkGenerator(
                            WatermarkGeneratorSupplier.Context context) {
                        return new WatermarkGenerator<CEPLoginBean>() {
                            private long maxTimeStamp = Long.MIN_VALUE;
                            private long maxOutOfOrderness = 500L;

                            @Override
                            public void onEvent(CEPLoginBean event, long eventTimestamp,
                                                WatermarkOutput output) {
                                maxTimeStamp = Math.max(maxTimeStamp, event.getTimestamp());
                            }


                            @Override
                            public void onPeriodicEmit(WatermarkOutput output) {
                                output.emitWatermark(new Watermark(maxTimeStamp - maxOutOfOrderness));
                            }
                        };
                    }
                }.withTimestampAssigner(((element, recordTimestamp) -> element.getTimestamp())));
        KeyedStream<CEPLoginBean, Long> keyedStream = watermarks.keyBy(value -> value.getId());
//        KeyedStream<CEPLoginBean, Long> keyedStream = data.keyBy(value -> value.getId());

        // 定义恶意登录的pattern
        // 用户在短时间内连着登录失败多次
        Pattern<CEPLoginBean, CEPLoginBean> pattern = Pattern
                .<CEPLoginBean>begin("start").where(new IterativeCondition<CEPLoginBean>() {
                    @Override
                    public boolean filter(CEPLoginBean value, Context<CEPLoginBean> ctx) throws Exception {
                        return value.getState().equals("fail");
                    }
                })
                .next("next").where(new IterativeCondition<CEPLoginBean>() {
                    @Override
                    public boolean filter(CEPLoginBean value, Context<CEPLoginBean> ctx) throws Exception {
                        return value.getState().equals("fail");
                    }
                })
                .within(Time.seconds(5));

        PatternStream<CEPLoginBean> patternStream = CEP.pattern(keyedStream, pattern);
        SingleOutputStreamOperator<String> process = patternStream.process(
                new PatternProcessFunction<CEPLoginBean, String>() {
                    @Override
                    public void processMatch(Map<String, List<CEPLoginBean>> match, Context ctx,
                                             Collector<String> out) throws Exception {
                        System.out.println(match);
                        List<CEPLoginBean> start = match.get("start");
                        List<CEPLoginBean> next = match.get("next");
                        String res = "start:" + start + "...next:" + next;
                        out.collect(res + start.get(0).getId());
                    }
                });

        process.print();

        env.execute();
    }
}
